package com.smsc.headend.task.engine.manager;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.text.csv.CsvUtil;
import cn.hutool.core.text.csv.CsvWriter;
import cn.hutool.core.util.*;
import cn.hutool.json.JSONUtil;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Lists;
import com.jcraft.jsch.SftpException;
import com.smsc.headend.common.enums.DeviceType;
import com.smsc.headend.common.utils.*;
import com.smsc.headend.module.asset.entity.Dcu;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.common.AppName;
import com.smsc.headend.module.common.entity.ScalerUnit;
import com.smsc.headend.module.data.entity.MapDdsMdmWf;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.data.enums.CollectionSource;
import com.smsc.headend.module.event.entity.*;
import com.smsc.headend.module.mdm.dds.consts.DdsConst;
import com.smsc.headend.module.mdm.dds.dto.MdmAnonyTerminalDto;
import com.smsc.headend.module.mdm.dds.dto.MdmDataDto;
import com.smsc.headend.module.mdm.dds.dto.MdmDeviceStatusDto;
import com.smsc.headend.module.mdm.dds.dto.MdmEventDataDto;
import com.smsc.headend.module.mdm.dds.entity.MapDdsMdmEventWf;
import com.smsc.headend.module.mdm.dds.enums.MdmEventDataItemEnum;
import com.smsc.headend.module.mdm.dds.enums.MdmSourceEnum;
import com.smsc.headend.module.mdm.dds.enums.MdmTableEnum;
import com.smsc.headend.module.sys.entity.Org;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.DeviceStatusDTO;
import com.smsc.headend.module.task.dto.TaskEventData;
import com.smsc.headend.module.template.loadprofile.entity.TemplateLoadProfileEntity;
import com.smsc.headend.module.vending.entity.MapVendingDataItem;
import com.smsc.headend.task.engine.ftp.config.SftpConfig;
import com.smsc.headend.task.engine.ftp.config.SftpUtils;
import com.smsc.headend.task.engine.mapper.MapDdsMdmWfMapper;
import com.smsc.headend.task.engine.po.csv.MsBasePo;
import com.smsc.headend.task.engine.po.csv.MsDataCsvPo;
import com.smsc.headend.task.engine.po.csv.MsEventCsvPo;
import com.smsc.headend.task.engine.po.xml.*;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.dds.MapDdsMdmEventWfService;
import com.smsc.headend.task.engine.service.dds.MapDdsMdmWfService;
import com.smsc.headend.task.engine.service.dds.OrgService;
import com.smsc.headend.task.engine.service.push.EventService;
import com.smsc.headend.task.engine.utils.DdsXmlGenerateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.list.UnmodifiableList;
import org.apache.commons.lang.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.util.Pair;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.smsc.headend.module.mdm.dds.consts.DdsConst.*;

@Slf4j
@Component
public class DdsFilterMdmDataManager {

    public static final String TXT = ".txt";
    public static final String TMP = ".tmp";

    private static final String CSV = ".csv";
    private static final String LOCK = ".lock";
    private static final String DAILY = "01";
    private static final String MONTHLY = "02";
    private static final String YEARLY = "03";
    private static final String EVENT = "04";
    private static final String UNDER_LINE = "_";

    public static final String METER_NO = "Meter No.";
    public static final String DATE_TIME = "Date Time";
    public static final String EVEN_CODE = "Even Code";
    public static final String EVENT_OCCUR_TIME = "Event Occur Time";

    public static final String YYYYMMDD = "yyyyMMdd";
    public static final String YYYYMM = "yyyyMM";

    public static final Long MIN_LP = 1020070010L;
    public static final Long DAY_LP = 1020070002L;
    public static final Long MONTH_LP = 1020070001L;
    public static final Long YEARLY_LP = 0L;
    public static final String MULTISPEAK_MIN = "Load_Profile";
    public static final String MULTISPEAK_DAY = "Daily_Profile";
    public static final String MULTISPEAK_MONTH = "Monthly_Profile";
    public static final String MULTISPEAK_EVENTS = "Events";
    public static final String MULTISPEAK_MIN_FOLDER = "Load-profile";
    public static final String MULTISPEAK_DAY_FOLDER = "Daily-profile";
    public static final String MULTISPEAK_MONTH_FOLDER = "Monthly-profile";
    public static final String MULTISPEAK_EVENTS_FOLDER = "Event-profile";
    public static final String HEAD_END_EXTERNAL_ID = "WASION-HES";

    public static final String EIP_FILE_END = "xml_EIP_eMeter";
    public static final String EIP_EVENTS = "Events";
    public static final String EIP_METER_READS = "MeterReads";
    public static final String EIP_METER_READS_MIN = "ElecCode";
    public static final String EIP_METER_READS_DAY = "ElecCodeD";
    public static final String EIP_METER_READS_MON = "ElecCodeM";
    public static final String NULL_STR = "NULL";


    @Autowired
    RedisUtils redisUtils;

    @Autowired
    SftpConfig sftpConfig;

    @Value("${sftp.localDir}")
    private String filePath;

    @Value("${smsc.asset.meter.single-phase-model:3002}")
    private String singlePhaseModel;

    @Autowired
    AssetService assetService;

    @Autowired
    SftpUtils sftpUtils;
    @Autowired
    MapDdsMdmWfService mapDdsMdmWfService;
    @Autowired
    MapDdsMdmEventWfService mapDdsMdmEventWfService;
    @Autowired
    FeignAssetManagementService feignAssetManagementService;
    @Autowired
    MapDdsMdmWfMapper mapDdsMdmWfMapper;
    @Autowired
    DdsXmlGenerateUtil ddsXmlGenerateUtil;
    @Autowired
    OrgService orgService;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    EventService eventService;

    private Map<String, MapDdsMdmWf> mappingDataRules = new HashMap<>(100);

    private Map<String, MapDdsMdmWf> mappingDataRulesCsv = new HashMap<>(100);

    private final Map<String, String> mappingEventRules = new HashMap<>(256);

    private final List<String> ignoreMappingRuleList = new ArrayList<>();
    /**
     * 以下几类数据才上来需要马上DDS到MDM--wf98定制
     */
    private static final List<String> FILE_ID_LIST_TO_INSTANT_UPLOAD = Arrays.asList(
            String.valueOf(DdsConst.DDS_HES_TERMINAL_STATE_TABLE_ID),
            String.valueOf(DdsConst.DDS_HES_ANONY_TERMINAL_TABLE_ID),
            String.valueOf(DdsConst.DDS_HES_METER_STATE_TABLE_ID),
            String.valueOf(DdsConst.DDS_RD_TERM_ALARM_TABLE_ID));

    private static final Integer EVENT_FLAG_OCCUR = 1;

    ConcurrentHashMap<Integer, Pair<String, Long>> ddsFileMap = new ConcurrentHashMap();

    ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();

    ReentrantReadWriteLock.ReadLock fileNameReadLock = reentrantReadWriteLock.readLock();

    ReentrantReadWriteLock.WriteLock fileNameWriteLock = reentrantReadWriteLock.writeLock();

    LoadingCache<String, ScalerUnit> subDataItemScaleUnit = CacheBuilder.newBuilder()
            //设置缓存容器的初始容量为10
            .initialCapacity(100)
            // maximumSize 设置缓存大小
            .maximumSize(65535)
            //设置并发级别为20，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(20)
            // expireAfterWrite设置写缓存后61秒钟过期
            .expireAfterWrite(60, TimeUnit.MINUTES)
            .build(new CacheLoader<String, ScalerUnit>() {
                @Override
                public ScalerUnit load(String modelIdDataModuleInfo) throws Exception {
                    String[] infos = modelIdDataModuleInfo.split(":");
                    ScalerUnit scalerUnitByModelIdSubDataItemId = feignAssetManagementService.getScalerUnitByModelIdSubDataItemId(Convert.toLong(infos[0]), Convert.toLong(infos[1]));

                    ScalerUnit scalerUnit = scalerUnitByModelIdSubDataItemId;
                    if (scalerUnitByModelIdSubDataItemId == null) {
                        scalerUnit = new ScalerUnit();
                        scalerUnit.setSubDataItemId(-1L);

                    }
                    return scalerUnit;
                }
            });

    LoadingCache<String, String> subDataItemCoverRate = CacheBuilder.newBuilder()
            //设置缓存容器的初始容量为10
            .initialCapacity(100)
            // maximumSize 设置缓存大小
            .maximumSize(65535)
            //设置并发级别为20，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(20)
            // expireAfterWrite设置写缓存后61秒钟过期
            .expireAfterWrite(60, TimeUnit.MINUTES)
            .build(new CacheLoader<String, String>() {
                @Override
                public String load(String modelIdDataModuleInfo) throws Exception {
                    String[] infos = modelIdDataModuleInfo.split(":");
                    String coverRate = feignAssetManagementService.getCoverRate(infos[2], Convert.toLong(infos[1]), Convert.toLong(infos[0]));
                    return coverRate == null ? NULL_STR : coverRate;
                }
            });


    LoadingCache<String, String> appUnitName = CacheBuilder.newBuilder()
            //设置缓存容器的初始容量为10
            .initialCapacity(100)
            // maximumSize 设置缓存大小
            .maximumSize(65535)
            //设置并发级别为20，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(200)
            // expireAfterWrite设置写缓存后61秒钟过期
            .expireAfterWrite(60, TimeUnit.MINUTES)
            .build(new CacheLoader<String, String>() {
                @Override
                public String load(String modelIdDataModuleInfo) throws Exception {
                    String[] infos = modelIdDataModuleInfo.split(":");
                    String subDataItemId = infos[0];
                    String appName = infos[1];
                    String coverRate = feignAssetManagementService.getAppConfigUnitName(appName, Convert.toLong(subDataItemId));
                    return coverRate == null ? NULL_STR : coverRate;
                }
            });

    List<Integer> reverseCoverRateUnitId = UnmodifiableList.unmodifiableList(Arrays.asList(27, 28, 29, 30, 31, 32));


    public String getMDMValue(Long modelId, Long subDataItemId, String value) {
        //Start: MDM要求所有时间格式以时间戳的形式返回.此处特殊处理
        String timestamp = null;
        if (StringUtils.isEmpty(value)) {
            return null;
        }
        if (!StringUtils.isEmpty(value) && value.contains("-") && value.contains(":")) {
            try {
                timestamp = String.valueOf(DateUtils.parseDate(value, new String[]{"yyyy-MM-dd HH:mm:ss"}).getTime() / 1000);
            } catch (Exception e) {
            }
        }
        if (!StringUtils.isEmpty(timestamp)) {
            return timestamp;
        }
        //End.
        if (!NumberUtil.isNumber(value)) {
            return value;
        }
        BigDecimal val = new BigDecimal(value);
        String key = modelId + ":" + subDataItemId + ":" + AppName.MDM.toString();
        ScalerUnit scalerUnit = getScaleUnit(key);
        if (scalerUnit == null) {
            //获取单位失败，不传输到mdm
            return null;
        }

        if (scalerUnit.getSubDataItemId() == -1L) {
            //获取到单位为空， 返回原值
            return value;
        }

        val = val.multiply(new BigDecimal(Math.pow(10, scalerUnit.getScaler()))).setScale((int) Math.abs(3), BigDecimal.ROUND_HALF_UP);

        String coverRate = getCoverRate(key);
        if (coverRate == null) {
            //获取量纲失败，不传输到mdm
            return null;
        }

        if (!NULL_STR.equalsIgnoreCase(coverRate) && !"ascii".equalsIgnoreCase(coverRate) && NumberUtil.isNumber(coverRate)) {
            if (reverseCoverRateUnitId.contains(scalerUnit.getUnit())) {
                Double rate = 1 / Double.parseDouble(coverRate);
                coverRate = rate.toString();
            }
            val = val.multiply(new BigDecimal(coverRate)).setScale((int) Math.abs(3), BigDecimal.ROUND_HALF_UP);
        }
        return val.toString();
    }

    public ScalerUnit getScaleUnit(String key) {
        ScalerUnit scalerUnit = subDataItemScaleUnit.getIfPresent(key);
        if (scalerUnit == null) {
            try {
                scalerUnit = subDataItemScaleUnit.get(key);
            } catch (Exception e) {
                log.error("getScaleUnitFailed,key:{},exception:", key, e);
                return null;
            }
        }
        return scalerUnit;
    }

    public String getAppConfigUnitName(String key) {
        String coverRate = appUnitName.getIfPresent(key);
        if (coverRate == null) {
            try {
                coverRate = appUnitName.get(key);
            } catch (Exception e) {
                log.error("getAppConfigUnitName,key:{},exception:", key, e);
                return null;
            }
        }
        return coverRate;
    }

    public String getCoverRate(String key) {
        String coverRate = subDataItemCoverRate.getIfPresent(key);
        if (coverRate == null) {
            try {
                coverRate = subDataItemCoverRate.get(key);
            } catch (Exception e) {
                log.error("getCoverRateFailed,key:{},exception:", key, e);
                return null;
            }
        }
        return coverRate;
    }


    /**
     * 下面几项数据需要加入‘曲线状态’至DDS文件，与梁立沟通，数据来源为rdMeterReadsLp的validity，
     * 与罗腾飞沟通，相同曲线的数据只取一次，故放任意子项对应的曲线状态均可。
     */
    private final List<Integer> mappingRuleWithProfileStatus = Arrays.asList(
            DDS_RD_ELEC_CODE_TABLE_ID
            , DDS_RD_PVC2_I_TABLE_ID
            , DDS_RD_PVC2_U_TABLE_ID
            , DDS_RD_PVC2_P_TABLE_ID
            , DDS_RD_PVC2_Q_TABLE_ID
            , DDS_RD_PVC2_COS_TABLE_ID
            , DDS_RD_PVC2_S_TABLE_ID);

    @PostConstruct
    public void initMappingRule() {
        //dds 数据配置(TXT)
        mappingDataRules = getMappingDataRules(1);
        //dds 数据配置(CSV)
        mappingDataRulesCsv = getMappingDataRules(2);
        ignoreMappingRuleList.add("1020070010:1020030122102");

        //dds 事件配置
        List<MapDdsMdmEventWf> mapDdsMdmEventWfList = mapDdsMdmEventWfService.list();
        mapDdsMdmEventWfList.stream().forEach(mapDdsMdmEventWf -> {
            Long eventTypeId = mapDdsMdmEventWf.getEventType();
            String mdmDataItemId = mapDdsMdmEventWf.getMdmEventDataItemId();
            String redisMapKey = new StringBuffer().append(eventTypeId).toString();
            mappingEventRules.put(RedisKeys.getMdmDataTypeId(redisMapKey), mdmDataItemId);
        });
    }

    public Map<String, MapDdsMdmWf> getMappingDataRules(Integer fileType) {
        Map<String, MapDdsMdmWf> map = new HashMap<>();
        //只获取TXT文件格式的配置
        List<MapDdsMdmWf> mapRelationList = mapDdsMdmWfService.getMapDdsMdmWfByFileType(fileType);
        mapRelationList.stream().forEach(mapRule -> {
            Long subDataItemId = mapRule.getSubDataItemId();
            Long lpDataItemId = mapRule.getLpDataItemId();
            String redisMapKey = new StringBuffer().append(lpDataItemId).append(":").append(subDataItemId).toString();
            map.put(RedisKeys.getMdmMapRule(redisMapKey), mapRule);
        });
        return map;
    }

    public void transformMeterReadResultForMdm(List<RdMeterReadsLp> taskReadDatas) {
        if (CollectionUtils.isEmpty(taskReadDatas)) {
            return;
        }
        log.debug("dds file generate:  lpReadSize={}", taskReadDatas.size());
        List<String> noMapDataItem = new ArrayList<>();
        //key： List ["lpDataItemId", "MapDdsMdmWfId" , “meterId”, “tv”]
        Map<List<Object>, List<RdMeterReadsLp>> groupedRdMeterReads = taskReadDatas.stream().collect(Collectors.groupingBy(readsLp -> {
                    String redisKey = new StringBuffer().append(readsLp.getLpDataItemId()).append(":").append(readsLp.getDataItemId()).toString();
                    MapDdsMdmWf mapDdsMdmWf = getMapDdsMdmWf(redisKey);
                    if (mapDdsMdmWf == null && !noMapDataItem.contains(redisKey)) {
                        noMapDataItem.add(redisKey);
                    }
                    return Arrays.asList(readsLp.getLpDataItemId(), mapDdsMdmWf == null ? -1 : mapDdsMdmWf.getMdmDataTypeId(), readsLp.getMeterId(), readsLp.getTv());
                }
        ));
        if (!CollectionUtil.isEmpty(noMapDataItem)) {
            log.info("dds file generate:  no MDM map config dataItem={}", JSONUtil.toJsonStr(noMapDataItem));
        }

        Map<String, List<String>> fileLineMap = new HashMap<>();
        for (Map.Entry<List<Object>, List<RdMeterReadsLp>> lpReadInfoEntry : groupedRdMeterReads.entrySet()) {
            List<Object> lpReadInfo = lpReadInfoEntry.getKey();
            Long lpDataItemId = Convert.toLong(lpReadInfo.get(0));
            if (lpDataItemId != 1020070010L) {
                log.info("grouped dds data: lpDataItemId={}, mdmDataTypeId={}, meterId={}, tv={}", lpDataItemId, lpReadInfo.get(1), lpReadInfo.get(2), lpReadInfo.get(3));
            }
            Integer mdmDataTypeId = Convert.toInt(lpReadInfo.get(1));
            if (mdmDataTypeId == -1) {
                continue;
            }
            Long meterId = Convert.toLong(lpReadInfo.get(2));
            Long tv = Convert.toLong(lpReadInfo.get(3));
            Meter meter = assetService.getMeterById(meterId);
            if (meter == null) {
                log.error("dds file Meter not exists. meterId={}", meterId);
                continue;
            }
            List<RdMeterReadsLp> readsLps = groupedRdMeterReads.get(lpReadInfo);
            List<MdmDataDto> mdmDataDtos = composeRdDataDto(readsLps, meter, lpDataItemId, tv);
            String fileName = getDdsFileNameByKey(mdmDataTypeId);
            log.debug("get FileName: {}", fileName);
            if (fileLineMap.containsKey(fileName)) {
                fileLineMap.get(fileName).addAll(mdmDataDtos.stream().map(mdmDataDto -> mdmDataDto.toMdmFormatString(mapDdsMdmWfService.getMaxSeqSizeByCode(mdmDataTypeId))).collect(Collectors.toList()));
            } else {
                fileLineMap.put(fileName, mdmDataDtos.stream().map(mdmDataDto -> mdmDataDto.toMdmFormatString(mapDdsMdmWfService.getMaxSeqSizeByCode(mdmDataTypeId))).collect(Collectors.toList()));
            }
        }

        String localFilePath = assetService.getSystemProperty("DDS_SFTP_LOCALDIR");
        if (StrUtil.isEmpty(localFilePath)) {
            localFilePath = filePath;
        }

        String finalLocalFilePath = localFilePath;
        fileLineMap.entrySet().stream().forEach(
                flMap -> {
                    String fileName = flMap.getKey();
                    StringBuilder absoluteFile = new StringBuilder().append(finalLocalFilePath).append(fileName);
                    File ddsFile = FileUtil.file(absoluteFile.append(TMP).toString());
                    FileUtil.appendUtf8Lines(flMap.getValue(), ddsFile);
                    long fileSize = FileUtil.size(ddsFile);
                    if (fileSize > 5 * 1024 * 1024 - 5000) {
                        String mdmDataTypeId = fileName.split("_")[1];
                        this.removeDdsFileNameKey(Convert.toInt(mdmDataTypeId));
                        String newFileName = StrUtil.concat(true, fileName, TXT);
                        FileUtil.rename(ddsFile, newFileName, false, true);
                        log.info("dds file generated：name={}, size={}", newFileName, fileSize);
                        //MDM文件格式 DATA_{mdmDataTypeId}_
                    }
                }
        );
    }

    /**
     * 转换年月日冻结曲线为Csv文件传递给DDS
     *
     * @param taskReadDatas
     */
    public void transformMeterReadResultForMdmCsv(List<RdMeterReadsLp> taskReadDatas) {
        if (CollectionUtils.isEmpty(taskReadDatas)) {
            return;
        }
        log.debug("dds file generate:  lpReadSize={}", taskReadDatas.size());
        List<String> noMapDataItem = new ArrayList<>();
        // key： List ["lpDataItemId", "MapDdsMdmWfId" , “meterId”, “tv”]
        Map<List<Object>, List<RdMeterReadsLp>> groupedRdMeterReads = taskReadDatas.stream().collect(Collectors.groupingBy(readsLp -> {
                    String redisKey = new StringBuffer().append(readsLp.getLpDataItemId()).append(":").append(readsLp.getDataItemId()).toString();
                    MapDdsMdmWf mapDdsMdmWf = getMapDdsMdmWfCsv(redisKey);
                    if (mapDdsMdmWf == null && !noMapDataItem.contains(redisKey)) {
                        noMapDataItem.add(redisKey);
                    }
                    return Arrays.asList(readsLp.getLpDataItemId(), mapDdsMdmWf == null ? -1 : mapDdsMdmWf.getMdmDataTypeId(), readsLp.getMeterId(), readsLp.getTv());
                }
        ));
        if (!CollectionUtil.isEmpty(noMapDataItem)) {
            log.info("dds file generate:  no MDM map config dataItem={}", JSONUtil.toJsonStr(noMapDataItem));
        }
        String localFilePath = assetService.getSystemProperty("DDS_SFTP_LOCALDIR");
        if (StrUtil.isEmpty(localFilePath)) {
            localFilePath = filePath;
        }
        createCsvFileAndUpload(groupedRdMeterReads, localFilePath);
    }

    /**
     * 转换年月日冻结曲线为xml文件传递给MDM
     *
     * @param taskReadDatas
     */
    public void transformMeterReadResultForOracleMdmXml(List<RdMeterReadsLp> taskReadDatas) {
        if (CollectionUtils.isEmpty(taskReadDatas)) {
            return;
        }
        log.debug("MultiSpeak dds xml file generate:  lpReadSize={}", taskReadDatas.size());
        Map<Long, List<RdMeterReadsLp>> groupedRdMeterReads = taskReadDatas
                .stream().collect(Collectors.groupingBy(readsLp -> readsLp.getLpDataItemId()));
        for (Map.Entry<Long, List<RdMeterReadsLp>> lpReadInfoEntry : groupedRdMeterReads.entrySet()) {
            Long lpDataItemId = lpReadInfoEntry.getKey();
            Map<Long, List<RdMeterReadsLp>> groupedByMeterRdMeterReads = groupedRdMeterReads.get(lpDataItemId)
                    .stream().collect(Collectors.groupingBy(readsLp -> readsLp.getMeterId()));
            String limitDeviceStr = assetService.getSystemProperty("DDS_MULTISPEAK_LIMIT_DEVICE");
            Integer limitNumber = Convert.toInt(limitDeviceStr, 10);
            List<Long> allMeterIds = new ArrayList<>(groupedByMeterRdMeterReads.keySet());
            int total = allMeterIds.size();
            int page = (int) Math.ceil((double) total / limitNumber);
            for (int pageNo = 0; pageNo < page; pageNo++) {
                List<Long> meterIds = CollUtil.page(pageNo, limitNumber, allMeterIds);
                DeviceList deviceList = transformMeterList(groupedByMeterRdMeterReads, meterIds, lpDataItemId);
                String localFilePath = assetService.getSystemProperty("DDS_MULTISPEAK_LOCAL_PATH");
                if (StrUtil.isEmpty(localFilePath)) {
                    localFilePath = filePath;
                }
                String fileName = "";
                String folder = "";
                String nowStr = DateUtil.format(new Date(), "yyyy-MM-dd_HH_mm_ss");
                if (lpDataItemId.equals(MIN_LP)) {
                    folder = MULTISPEAK_MIN_FOLDER;
                    fileName = MULTISPEAK_MIN + "_" + nowStr + "_" + (pageNo + 1) + ".xml";
                } else if (lpDataItemId.equals(DAY_LP)) {
                    folder = MULTISPEAK_DAY_FOLDER;
                    fileName = MULTISPEAK_DAY + "_" + nowStr + "_" + (pageNo + 1) + ".xml";
                } else if (lpDataItemId.equals(MONTH_LP)) {
                    folder = MULTISPEAK_MONTH_FOLDER;
                    fileName = MULTISPEAK_MONTH + "_" + nowStr + "_" + (pageNo + 1) + ".xml";
                }
                File file = FileUtil.file(localFilePath, folder, fileName);
                FileUtil.writeString(XmlUtils.beanToXml(deviceList), file, "UTF-8");
            }
        }
    }


    /**
     * 转换年月日冻结曲线为csv文件传递给MDM
     *
     * @param taskReadDatas
     */
    public void transformMeterReadResultForOracleMdmCsv(List<RdMeterReadsLp> taskReadDatas) {
        if (CollectionUtils.isEmpty(taskReadDatas)) {
            return;
        }
        log.debug("MultiSpeak dds csv file generate:  lpReadSize={}", taskReadDatas.size());
        Map<Long, List<RdMeterReadsLp>> groupedRdMeterReads = taskReadDatas
                .stream().collect(Collectors.groupingBy(readsLp -> readsLp.getLpDataItemId()));
        for (Map.Entry<Long, List<RdMeterReadsLp>> lpReadInfoEntry : groupedRdMeterReads.entrySet()) {
            Long lpDataItemId = lpReadInfoEntry.getKey();
            if (!lpDataItemId.equals(MIN_LP)) {
                continue;
            }
            Map<Long, List<RdMeterReadsLp>> groupedByMeterRdMeterReads = groupedRdMeterReads.get(lpDataItemId)
                    .stream().collect(Collectors.groupingBy(readsLp -> readsLp.getMeterId()));
            String limitDeviceStr = assetService.getSystemProperty("DDS_MULTISPEAK_LIMIT_DEVICE");
            Integer limitNumber = Convert.toInt(limitDeviceStr, 10);
            List<Long> allMeterIds = new ArrayList<>(groupedByMeterRdMeterReads.keySet());
            int total = allMeterIds.size();
            int page = (int) Math.ceil((double) total / limitNumber);
            for (int pageNo = 0; pageNo < page; pageNo++) {
                List<Long> meterIds = CollUtil.page(pageNo, limitNumber, allMeterIds);
                List<MsDataCsvPo> msDataCsvPos = transformCsvMeterList(groupedByMeterRdMeterReads, meterIds, lpDataItemId);
                if (CollUtil.isEmpty(msDataCsvPos)) {
                    continue;
                }
                String localFilePath = assetService.getSystemProperty("DDS_MULTISPEAK_LOCAL_PATH");
                if (StrUtil.isEmpty(localFilePath)) {
                    localFilePath = filePath;
                }
                String folder = MULTISPEAK_MIN_FOLDER;
                String fileName = HEAD_END_EXTERNAL_ID + SystemClock.now() + ".csv";
                File file = FileUtil.file(localFilePath, folder, fileName);
                CsvWriter writer = CsvUtil.getWriter(file, CharsetUtil.CHARSET_UTF_8);
                msDataCsvPos.stream().forEach(msDataCsvPo -> {
                    Map<String, Object> map = BeanUtil.beanToMap(msDataCsvPo.getMsBasePo());
                    writer.writeLine(ArrayUtil.addAll(Convert.toStrArray(map.values()), msDataCsvPo.getReadings()));
                });
                // write total records
                writer.writeLine(new String[]{"T", Convert.toStr(SystemClock.now() / 1000), Convert.toStr(msDataCsvPos.size())});
                writer.flush();
                writer.close();
            }
        }
    }

    /**
     * 转换年月日冻结曲线为xml文件传递给EIP
     *
     * @param taskReadDatas
     */
    public void transformMeterReadResultForEIPXml(List<RdMeterReadsLp> taskReadDatas) {
        if (CollectionUtils.isEmpty(taskReadDatas)) {
            return;
        }
        List<RdMeterReadsLp> taskReadDatasMin = new ArrayList<>();
        List<RdMeterReadsLp> taskReadDatasDay = new ArrayList<>();
        List<RdMeterReadsLp> taskReadDatasMon = new ArrayList<>();
        Map<String, List<RdMeterReadsLp>> meterReadsMap = new HashMap(16);
        taskReadDatas.forEach(taskReadData -> {
                    TemplateLoadProfileEntity templateLoadProfileEntity = feignAssetManagementService.getTemplateLoadProfileEntity(taskReadData.getMeterId(), taskReadData.getLpDataItemId());
                    if (null != templateLoadProfileEntity) {
                        String intervalType = templateLoadProfileEntity.getIntervalType();
                        if (TemplateLoadProfileEntity.IntervalType.MINUTELY.getType().equals(intervalType)) {
                            taskReadDatasMin.add(taskReadData);
                        }
                        if (TemplateLoadProfileEntity.IntervalType.DAILY.getType().equals(intervalType)) {
                            taskReadDatasDay.add(taskReadData);
                        }
                        if (TemplateLoadProfileEntity.IntervalType.MONTHLY.getType().equals(intervalType)) {
                            taskReadDatasMon.add(taskReadData);
                        }
                    }
                }

        );
        if (!CollUtil.isEmpty(taskReadDatasMin)) {
            meterReadsMap.put(EIP_METER_READS_MIN, taskReadDatasMin);
        }
        if (!CollUtil.isEmpty(taskReadDatasDay)) {
            meterReadsMap.put(EIP_METER_READS_DAY, taskReadDatasDay);
        }
        if (!CollUtil.isEmpty(taskReadDatasMon)) {
            meterReadsMap.put(EIP_METER_READS_MON, taskReadDatasMon);
        }
        for (String key : meterReadsMap.keySet()) {
            List<RdMeterReadsLp> readsLps = meterReadsMap.get(key);
            log.debug("EIP dds file {} generate:  lpReadSize={}", key, readsLps.size());
            String localFilePath = assetService.getSystemProperty("DDS_EIP_LOCAL_PATH");
            if (StrUtil.isEmpty(localFilePath)) {
                localFilePath = filePath;
            }
            String orgName = "NULL";
            Meter meter = assetService.getMeterById(readsLps.get(0).getMeterId());
            if (null != meter) {
                Org org = orgService.getById(meter.getOrgId());
                if (null != org) {
                    orgName = org.getOrgName();
                }
            }
            String nowStr = DateUtil.format(new Date(), "yyyyMMddHHmmssSSS");
            String folder = EIP_METER_READS;
            String fileName = key + "_" + nowStr + "_" + EIP_FILE_END + "_" + orgName + ".xml";
            File file = FileUtil.file(localFilePath, folder, fileName);
            String str = ddsXmlGenerateUtil.getDocumentLp(readsLps).toString();
            FileUtil.writeString(str, file, "UTF-8");
            log.debug("EIP Meter Reads File Name:" + fileName);
        }

    }

    private DeviceList transformMeterList(Map<Long, List<RdMeterReadsLp>> groupedByMeterRdMeterReads, List<Long> meterIds, Long lpDataItemId) {
        DeviceList deviceList = new DeviceList();
        meterIds.stream().forEach(meterId -> {
            Device device = new Device();
            Meter meter = assetService.getMeterById(meterId);
            device.setDeviceIdentifierNumber(meter.getCustomerSn());
            device.setHeadEndExternalId(HEAD_END_EXTERNAL_ID);
            List<RdMeterReadsLp> rdMeterReadsLps = groupedByMeterRdMeterReads.get(meterId);
            if (lpDataItemId.equals(MIN_LP)) {
                device.setInitialMeasurementDataList(transformMinInitialMeasurementDataList(meter, rdMeterReadsLps));
            } else if (lpDataItemId.equals(DAY_LP) || lpDataItemId.equals(MONTH_LP)) {
                device.setInitialMeasurementDataList(transformMonthlyOrDailyInitialMeasurementDataList(meter, rdMeterReadsLps, lpDataItemId));
            }
            deviceList.getDevice().add(device);
        });

        return deviceList;
    }

    private List<MsDataCsvPo> transformCsvMeterList(Map<Long, List<RdMeterReadsLp>> groupedByMeterRdMeterReads, List<Long> meterIds, Long lpDataItemId) {
        List<MsDataCsvPo> list = CollUtil.newArrayList();
        meterIds.stream().forEach(meterId -> {
            Meter meter = assetService.getMeterById(meterId);
            if (meter.getManageStatus() == Meter.MeterManageStatus.Managed.getCode()) {
                Integer interval = feignAssetManagementService.getProfileIntervalTimeByLpDataItemId(meterId, lpDataItemId);
                Map<Long, List<RdMeterReadsLp>> groupedByDataItemIdReads = groupedByMeterRdMeterReads.get(meterId)
                        .stream().collect(Collectors.groupingBy(readsLp -> readsLp.getDataItemId()));
                groupedByDataItemIdReads.forEach((dataItemId, readsLps) -> {
                    MapVendingDataItem multiSpeakDataItem = assetService.getMultispeakDataItem(dataItemId, MapVendingDataItem.Type.MULTISPEAK_UNIT.getCode(), lpDataItemId);
                    if (multiSpeakDataItem != null) {
                        MsDataCsvPo msDataCsvPo = new MsDataCsvPo();
                        MsBasePo msBasePo = new MsBasePo();
                        msDataCsvPo.setMsBasePo(msBasePo);
                        msBasePo.setSerialNumber(meter.getCustomerSn());
                        List<RdMeterReadsLp> sortList = CollUtil.sortByProperty(readsLps, "tv");
                        msBasePo.setTvStart(Convert.toStr(sortList.get(0).getTv() - interval));
                        msBasePo.setTvEnd(sortList.get(sortList.size() - 1).getTv().toString());
                        msBasePo.setInterval(Convert.toStr(interval));
                        msDataCsvPo.setReadings(transformReadingsByInterval(meter.getModelId(), interval, sortList));
                        msBasePo.setUom(multiSpeakDataItem.getVendingRef());
                        list.add(msDataCsvPo);
                    }
                });
            }
        });

        return list;
    }

    private String[] transformReadingsByInterval(Long modelId, Integer interval, List<RdMeterReadsLp> reads) {
        List<String> list = CollUtil.newArrayList();
        Long loopTv = null;
        for (int i = 0; i < reads.size(); i++) {
            RdMeterReadsLp rdMeterReadsLp = reads.get(i);
            Object[] unitValue = feignAssetManagementService.convertDataItemValWithUnit(modelId, rdMeterReadsLp.getDataItemId(), rdMeterReadsLp.getValue(), AppName.MDM.toString());
            if (i == 0) {
                loopTv = reads.get(0).getTv();
            } else {
                loopTv += interval;
                if ((rdMeterReadsLp.getTv() - loopTv) % interval != 0) {
                    continue;
                }
                while (loopTv < rdMeterReadsLp.getTv()) {
                    list.add("0:M");
                    loopTv += interval;
                }
            }
            list.add(getIfNullStr(unitValue[0].toString()));
        }

        return ArrayUtil.toArray(list, String.class);
    }

    private String getIfNullStr(String data) {
        if (StrUtil.isEmpty(data)) {
            return "0:M";
        }

        return data;
    }

    private InitialMeasurementDataList transformMinInitialMeasurementDataList(Meter meter, List<RdMeterReadsLp> rdMeterReadsLps) {
        InitialMeasurementDataList initialMeasurementDataList = new InitialMeasurementDataList();
        Map<Long, List<RdMeterReadsLp>> dataItemGroup = rdMeterReadsLps.stream().collect(Collectors.groupingBy(RdMeterReadsLp::getDataItemId));
        for (Map.Entry<Long, List<RdMeterReadsLp>> dataItemEntry : dataItemGroup.entrySet()) {
            Long dataItemId = dataItemEntry.getKey();
            MapVendingDataItem multiSpeakDataItem = assetService.getMultispeakDataItem(dataItemId, MapVendingDataItem.Type.MULTISPEAK.getCode(), MIN_LP);
            if (multiSpeakDataItem == null) {
                continue;
            }
            InitialMeasurementData initialMeasurementData = new InitialMeasurementData();
            PreVEE preVEE = new PreVEE();
            List<RdMeterReadsLp> readsLps = dataItemGroup.get(dataItemId);
            List<RdMeterReadsLp> sortList = CollUtil.sortByProperty(readsLps, "tv");
            Long startTv = sortList.get(0).getTv() - 60 * 60;
            Long endTv = sortList.get(sortList.size() - 1).getTv();
            preVEE.setExternalUOM(multiSpeakDataItem.getVendingRef());
            preVEE.setStDt(UDateUtils.convertToXMLGregorianCalendar(new Date(startTv * 1000)));
            preVEE.getStDt().setFractionalSecond(null);
            preVEE.setEnDt(UDateUtils.convertToXMLGregorianCalendar(new Date(endTv * 1000)));
            preVEE.getEnDt().setFractionalSecond(null);
            Msrs msrs = new Msrs();
            preVEE.setMsrs(msrs);
            int num = 1;
            for (RdMeterReadsLp rdMeterReadsLp : sortList) {
                ML ml = new ML();
                ml.setS(BigInteger.valueOf(num++));
                ml.setDt(UDateUtils.convertToXMLGregorianCalendar(new Date(rdMeterReadsLp.getTv() * 1000)));
                Object[] unitValue = feignAssetManagementService.convertDataItemValWithUnit(meter.getModelId(), rdMeterReadsLp.getDataItemId(), rdMeterReadsLp.getValue(), AppName.MDM.toString());
                ml.setQ(Convert.toBigDecimal(unitValue[0]));
                msrs.getML().add(ml);
            }
            initialMeasurementData.setPreVEE(preVEE);
            initialMeasurementDataList.getInitialMeasurementData().add(initialMeasurementData);
        }

        return initialMeasurementDataList;
    }

    private InitialMeasurementDataList transformMonthlyOrDailyInitialMeasurementDataList(Meter meter, List<RdMeterReadsLp> rdMeterReadsLps, Long lpDataItemId) {
        InitialMeasurementDataList initialMeasurementDataList = new InitialMeasurementDataList();
        for (RdMeterReadsLp rdMeterReadsLp : rdMeterReadsLps) {
            MapVendingDataItem multiSpeakDataItem = assetService.getMultispeakDataItem(rdMeterReadsLp.getDataItemId(), MapVendingDataItem.Type.MULTISPEAK.getCode(), lpDataItemId);
            if (multiSpeakDataItem == null) {
                continue;
            }
            InitialMeasurementData initialMeasurementData = new InitialMeasurementData();
            PreVEE preVEE = new PreVEE();
            preVEE.setExternalUOM(multiSpeakDataItem.getVendingRef());
            preVEE.setEnDt(UDateUtils.convertToXMLGregorianCalendar(new Date(rdMeterReadsLp.getTv() * 1000)));
            preVEE.getEnDt().setFractionalSecond(null);
            Object[] unitValue = feignAssetManagementService.convertDataItemValWithUnit(meter.getModelId(), rdMeterReadsLp.getDataItemId(), rdMeterReadsLp.getValue(), AppName.MDM.toString());
            preVEE.setEnQty(Convert.toBigDecimal(unitValue[0]));
            initialMeasurementData.setPreVEE(preVEE);
            initialMeasurementDataList.getInitialMeasurementData().add(initialMeasurementData);
        }

        return initialMeasurementDataList;
    }

    private DeviceList transformEventListForOracleMdmXml(Map<Long, List<RdEventMeter>> groupedMap, List<Long> meterIds) {
        DeviceList deviceList = new DeviceList();
        meterIds.stream().forEach(meterId -> {
            Device device = new Device();
            Meter meter = assetService.getMeterById(meterId);
            device.setDeviceIdentifierNumber(meter.getCustomerSn());
            device.setHeadEndExternalId(HEAD_END_EXTERNAL_ID);
            DeviceEventList deviceEventList = new DeviceEventList();
            device.setDeviceEventList(deviceEventList);
            List<RdEventMeter> rdEvents = groupedMap.get(meterId);
            for (RdEventMeter rdEvent : rdEvents) {
                DeviceEvent deviceEvent = new DeviceEvent();
                MapEventAlarm mapEventAlarm = assetService.getMapEventAlarm(rdEvent.getEventType());
                if (mapEventAlarm != null) {
                    if ("A".equals(mapEventAlarm.getType())) {
                        sendAlarmToAdapter(meter.getCustomerSn(), rdEvent, mapEventAlarm);
                        continue;
                    }
                    deviceEvent.setExternalEventName(mapEventAlarm.getMapCode().toString());
                } else {
                    deviceEvent.setExternalEventName(rdEvent.getEventCode().toString());
                }
                deviceEvent.setEventDateTime(UDateUtils.convertToXMLGregorianCalendar(new Date(rdEvent.getTv() * 1000)));
                deviceEventList.getDeviceEvent().add(deviceEvent);
            }
            deviceList.getDevice().add(device);
        });

        return deviceList;
    }

    private List<MsEventCsvPo> transformEventListForOracleMdmCsv(Map<Long, List<RdEventMeter>> groupedMap, List<Long> meterIds) {
        List<MsEventCsvPo> list = CollUtil.newArrayList();
        meterIds.stream().forEach(meterId -> {
            Meter meter = assetService.getMeterById(meterId);
            if (meter.getManageStatus() == Meter.MeterManageStatus.Managed.getCode()) {
                List<RdEventMeter> rdEvents = groupedMap.get(meterId);
                for (RdEventMeter rdEvent : rdEvents) {
                    MsEventCsvPo msEventCsvPo = new MsEventCsvPo();
                    msEventCsvPo.setSerialNumber(meter.getCustomerSn());
                    msEventCsvPo.setTv(rdEvent.getTv().toString());
                    if (StrUtil.isEmpty(rdEvent.getEventName())) {
                        String eventName = eventService.getEventName(Convert.toLong(rdEvent.getEventType()));
                        rdEvent.setEventName(eventName);
                    } else {
                        msEventCsvPo.setEventName(rdEvent.getEventName());
                    }
                    list.add(msEventCsvPo);
                }
            }
        });

        return list;
    }

    private void sendAlarmToAdapter(String serialNum, RdEventMeter rdEvent, MapEventAlarm mapEventAlarm) {
        D1DeviceEventSeeder d1DeviceEventSeeder = new D1DeviceEventSeeder();
        d1DeviceEventSeeder.setExternalSenderId(HEAD_END_EXTERNAL_ID);
        d1DeviceEventSeeder.setExternalEventName(mapEventAlarm.getMapCode().toString());
        d1DeviceEventSeeder.setEventDateTime(UDateUtils.convertToXMLGregorianCalendar(new Date(rdEvent.getTv() * 1000)));
        d1DeviceEventSeeder.setExecutionMethod("D1RT");
        d1DeviceEventSeeder.setRawEventInformation("");
        EventInformation eventInformation = new EventInformation();
        eventInformation.setExternalEventReason("REAL_TIME");
        d1DeviceEventSeeder.setEventInformation(eventInformation);
        d1DeviceEventSeeder.setDeviceIdentifierNumber(serialNum);
        kafkaTemplate.send(TaskKafkaTopic.C2M_ALARM_TO_ADAPTER, XmlUtils.beanToXml(d1DeviceEventSeeder).getBytes(StandardCharsets.UTF_8));
    }

    private List<MdmDataDto> composeRdDataDto(List<RdMeterReadsLp> readsLps, Meter meter, Long lpDataItemId, Long tv) {
        Map<Integer, MdmDataDto> feeTypeMap = new HashMap<>();
        Integer source;
        //默认validity
        Integer validity;
        for (RdMeterReadsLp readsLp : readsLps) {
            source = readsLp.getSrc();
            validity = readsLp.getValidity();
            String lpIdAndSubId = getLpIdAndSubId(lpDataItemId, readsLp.getDataItemId());
            if (ignoreMappingRuleList.contains(lpIdAndSubId)) {
                continue;
            }
            MapDdsMdmWf mapDdsMdmWf = getMapDdsMdmWf(lpIdAndSubId);
            if (mapDdsMdmWf == null) {
                log.error("MapDdsMdmWf: not exist, key={}", lpDataItemId);
                continue;
            }
            //单相表获取识别的相位传输
            if (isSinglePhaseModel(meter.getModelId())) {
                if ("B".equalsIgnoreCase(meter.getPhase()) && mapDdsMdmWf.getPhaseBSubDataItemId() != null) {
                    mapDdsMdmWf = getMapDdsMdmWf(getLpIdAndSubId(lpDataItemId, mapDdsMdmWf.getPhaseBSubDataItemId()));
                } else if ("C".equalsIgnoreCase(meter.getPhase()) && mapDdsMdmWf.getPhaseCSubDataItemId() != null) {
                    mapDdsMdmWf = getMapDdsMdmWf(getLpIdAndSubId(lpDataItemId, mapDdsMdmWf.getPhaseCSubDataItemId()));
                }
            }
            Integer feeType = Integer.valueOf(mapDdsMdmWf.getMdmDataAttr());
            MdmDataDto mdmDataDto;
            if (!feeTypeMap.containsKey(feeType)) {
                mdmDataDto = new MdmDataDto();
                mdmDataDto.setId(String.valueOf(meter.getCustomerSn()));
                mdmDataDto.setTv(tv);
                mdmDataDto.setValueMap(new HashMap<>());
                mdmDataDto.setSrc(getMdmSource(source));
                mdmDataDto.setValidity(validity);
                mdmDataDto.setFeeType(feeType);
                feeTypeMap.put(feeType, mdmDataDto);
            } else {
                mdmDataDto = feeTypeMap.get(feeType);
            }

            if (mdmDataDto.getTvUpdate() == null) {
                mdmDataDto.setTvUpdate(readsLp.getTvUpdate());
            }
            if (mdmDataDto.getProfileStatus() == null && mappingRuleWithProfileStatus.contains(mapDdsMdmWf.getMdmDataTypeId())) {
                mdmDataDto.setProfileStatus(readsLp.getValidity());
            }
            String value = getMDMValue(meter.getModelId(), readsLp.getDataItemId(), readsLp.getValue());
            if (value == null) {
                log.error("MDM convert data null, {}", JSONUtil.toJsonStr(readsLp));
                continue;
            }
            mdmDataDto.getValueMap().put(mapDdsMdmWf.getMdmDataSeqIndex(), value);
        }

        return Lists.newArrayList(feeTypeMap.values());
    }

    /**
     * MDM
     * 1	定抄
     * 2	补抄
     * 3	随抄
     * 4	设备上报
     * 11	接口(外部系统)
     *
     * @param source
     * @return
     */
    private String getMdmSource(Integer source) {
        switch (CollectionSource.valueOf(source)) {
            case ARR:
            case MARR:
                return "2";
            case SCH:
                return "1";
            case ODR:
                return "3";
            case FTP:
                return "4";
            default:
                return "1";
        }
    }

    private String getLpIdAndSubId(Long lpDataItemId, Long dataItemId) {
        return new StringBuffer().append(lpDataItemId).append(":").append(dataItemId).toString();
    }

    private boolean isSinglePhaseModel(Long modelId) {
        if (StringUtils.isEmpty(singlePhaseModel)) {
            return false;
        }
        List<Long> modelIds = Stream.of(singlePhaseModel.split(",")).map(Long::valueOf).collect(Collectors.toList());
        return modelIds.contains(modelId);
    }

    private synchronized String getDdsFileNameByKey(Integer mdmDataTypeId) {
        fileNameReadLock.lock();
        try {
            if (ddsFileMap.containsKey(mdmDataTypeId)) {
                return ddsFileMap.get(mdmDataTypeId).getFirst();
            }
            fileNameReadLock.unlock();
            fileNameWriteLock.lock();
            try {
                fileNameReadLock.lock();
                if (ddsFileMap.containsKey(mdmDataTypeId)) {
                    return ddsFileMap.get(mdmDataTypeId).getFirst();
                }
                StringBuilder fileNameBuilder = new StringBuilder();
                fileNameBuilder.append("DATA_").append(mdmDataTypeId)
                        .append("_").append(MdmTableEnum.getTextByCode(mdmDataTypeId))
                        .append("_").append(SystemClock.now() / 1000).append("_21_").append(IdUtil.simpleUUID());

                ddsFileMap.put(mdmDataTypeId, Pair.of(fileNameBuilder.toString(), SystemClock.now()));
                return fileNameBuilder.toString();
            } finally {
                fileNameWriteLock.unlock();
            }
        } finally {
            fileNameReadLock.unlock();
        }
    }

    private void removeDdsFileNameKey(Integer mdmDataTypeId) {
        fileNameReadLock.lock();
        try {
            if (!ddsFileMap.containsKey(mdmDataTypeId)) {
                return;
            }
            fileNameReadLock.unlock();
            fileNameWriteLock.lock();
            try {
                fileNameReadLock.lock();
                if (!ddsFileMap.containsKey(mdmDataTypeId)) {
                    return;
                }
                ddsFileMap.remove(mdmDataTypeId);
            } finally {
                fileNameWriteLock.unlock();
            }
        } finally {
            fileNameReadLock.unlock();
        }
    }


    private MapDdsMdmWf getMapDdsMdmWf(String redisKey) {
        return mappingDataRules.get(RedisKeys.getMdmMapRule(redisKey));
    }

    private MapDdsMdmWf getMapDdsMdmWfCsv(String redisKey) {
        return mappingDataRulesCsv.get(RedisKeys.getMdmMapRule(redisKey));
    }

    @Async
    public void generateTxtFilesAndUploadTheSelectedType(Map<Integer, List<String>> mdmDataList) {
        Iterator<Map.Entry<Integer, List<String>>> iterator = mdmDataList.entrySet().iterator();
        String localFilePath = assetService.getSystemProperty("DDS_SFTP_LOCALDIR");
        if (StrUtil.isEmpty(localFilePath)) {
            localFilePath = filePath;
        }
        while (iterator.hasNext()) {
            Map.Entry<Integer, List<String>> entry = iterator.next();
            List<String> results = entry.getValue();
            // format is DATA_22100_RD_ELEC_CODE_1489200269_21_c96c12601a67e14e4e19ab0126508c9ba9f6.txt
            StringBuilder name = new StringBuilder();
            String key = String.valueOf(entry.getKey());
            name.append("DATA_").append(key)
                    .append("_").append(MdmTableEnum.getTextByCode(Integer.parseInt(key)))
                    .append("_").append(System.currentTimeMillis() / 1000).append("_21_").append(IdUtil.simpleUUID());
            File file = TxtFileUtil.saveFile(localFilePath, name.toString(), results);
            judgeFileTypeAndUploadInstantly(key, file);
        }
    }

    /**
     * EVEN数据生成CSV文件并上传
     *
     * @param mdmDataList
     */
    @Async
    public void generateCsvFilesAndUploadTheSelectedType(Map<Integer, List<String>> mdmDataList) {
        Iterator<Map.Entry<Integer, List<String>>> iterator = mdmDataList.entrySet().iterator();
        String localFilePath = assetService.getSystemProperty("DDS_SFTP_LOCALDIR");
        if (StrUtil.isEmpty(localFilePath)) {
            localFilePath = filePath;
        }
        while (iterator.hasNext()) {
            Map.Entry<Integer, List<String>> entry = iterator.next();
            List<String> results = entry.getValue();
            createCsvEvenFileAndUpload(results, localFilePath);
        }
    }

    /**
     * EVEN数据生成XML文件并上传Oracle mdm
     *
     * @param taskEventDatas
     */
    public void generateXmlFilesForOracleMdm(List<TaskEventData> taskEventDatas) {
        Map<Long, List<RdEventMeter>> groupedMap = taskEventDatas.stream().filter(o -> !CollectionUtils.isEmpty(o.getRdEventMeter()))
                .map(TaskEventData::getRdEventMeter).flatMap(Collection::stream).collect(Collectors.groupingBy(RdEventMeter::getMeterId));
        List<Long> allMeterIds = new ArrayList<>(groupedMap.keySet());
        int total = allMeterIds.size();
        String limitDeviceStr = assetService.getSystemProperty("DDS_MULTISPEAK_LIMIT_DEVICE");
        Integer limitNumber = Convert.toInt(limitDeviceStr, 10);
        int page = (int) Math.ceil((double) total / limitNumber);
        for (int pageNo = 0; pageNo < page; pageNo++) {
            List<Long> meterIds = CollUtil.page(pageNo, limitNumber, allMeterIds);
            DeviceList deviceList = transformEventListForOracleMdmXml(groupedMap, meterIds);
            String localFilePath = assetService.getSystemProperty("DDS_MULTISPEAK_LOCAL_PATH");
            if (StrUtil.isEmpty(localFilePath)) {
                localFilePath = filePath;
            }
            String nowStr = DateUtil.format(new Date(), "yyyy-MM-dd_HH_mm_ss");
            String folder = MULTISPEAK_EVENTS_FOLDER;
            String fileName = MULTISPEAK_EVENTS + "_" + nowStr + "_" + (pageNo + 1) + ".xml";
            File file = FileUtil.file(localFilePath, folder, fileName);
            FileUtil.writeString(XmlUtils.beanToXml(deviceList), file, "UTF-8");
        }
    }

    /**
     * EVEN数据生成CSV文件并上传Oracle mdm
     *
     * @param taskEventDatas
     */
    public void generateCsvFilesForOracleMdm(List<TaskEventData> taskEventDatas) {
        Map<Long, List<RdEventMeter>> groupedMap = taskEventDatas.stream().filter(o -> !CollectionUtils.isEmpty(o.getRdEventMeter()))
                .map(TaskEventData::getRdEventMeter).flatMap(Collection::stream).collect(Collectors.groupingBy(RdEventMeter::getMeterId));
        List<Long> allMeterIds = new ArrayList<>(groupedMap.keySet());
        int total = allMeterIds.size();
        String limitDeviceStr = assetService.getSystemProperty("DDS_MULTISPEAK_LIMIT_DEVICE");
        Integer limitNumber = Convert.toInt(limitDeviceStr, 10);
        int page = (int) Math.ceil((double) total / limitNumber);
        for (int pageNo = 0; pageNo < page; pageNo++) {
            List<Long> meterIds = CollUtil.page(pageNo, limitNumber, allMeterIds);
            List<MsEventCsvPo> msEventCsvPos = transformEventListForOracleMdmCsv(groupedMap, meterIds);
            if(CollUtil.isEmpty(msEventCsvPos)) {
                return;
            }
            String localFilePath = assetService.getSystemProperty("DDS_MULTISPEAK_LOCAL_PATH");
            if (StrUtil.isEmpty(localFilePath)) {
                localFilePath = filePath;
            }
            String folder = MULTISPEAK_EVENTS_FOLDER;
            String fileName = HEAD_END_EXTERNAL_ID + SystemClock.now() + ".csv";
            File file = FileUtil.file(localFilePath, folder, fileName);
            CsvWriter writer = CsvUtil.getWriter(file, CharsetUtil.CHARSET_UTF_8);
            msEventCsvPos.stream().forEach(msEventCsvPo -> {
                Map<String, Object> map = BeanUtil.beanToMap(msEventCsvPo);
                writer.writeLine(Convert.toStrArray(map.values()));
            });
            // write total records
            writer.writeLine(new String[]{"T", Convert.toStr(SystemClock.now() / 1000), Convert.toStr(msEventCsvPos.size())});
            writer.flush();
            writer.close();
        }
    }

    /**
     * EVEN数据生成XML文件并上传EIP
     *
     * @param taskEventDatas
     */
    public void generateXmlEventFilesForEip(List<TaskEventData> taskEventDatas) {
        if (CollectionUtils.isEmpty(taskEventDatas)) {
            return;
        }
        log.debug("EIP dds file generate:  lpReadSize={}", taskEventDatas.size());
        String localFilePath = assetService.getSystemProperty("DDS_EIP_LOCAL_PATH");
        if (StrUtil.isEmpty(localFilePath)) {
            localFilePath = filePath;
        }
        String orgName = "NULL";
        TaskEventData taskEventData = taskEventDatas.get(0);
        if (DeviceType.METER.getId().equals(taskEventData.getDeviceType())) {
            Meter meter = assetService.getMeterById(taskEventData.getDeviceId());
            if (null != meter) {
                Org org = orgService.getById(meter.getOrgId());
                if (null != org) {
                    orgName = org.getOrgName();
                }
            }
        }
        if (DeviceType.DCU.getId().equals(taskEventData.getDeviceType())) {
            Dcu dcu = assetService.getDcuById(taskEventData.getDeviceId());
            if (null != dcu) {
                Org org = orgService.getById(dcu.getOrgId());
                if (null != org) {
                    orgName = org.getOrgName();
                }
            }
        }
        String nowStr = DateUtil.format(new Date(), "yyyyMMddHHmmssSSS");
        String folder = EIP_EVENTS;
        String fileName = EIP_EVENTS + "_" + nowStr + "_" + EIP_FILE_END + "_" + orgName + ".xml";
        File file = FileUtil.file(localFilePath, folder, fileName);
        String str = ddsXmlGenerateUtil.getDocumentEvent(taskEventDatas).toString();
        FileUtil.writeString(str, file, "UTF-8");
    }

    private void judgeFileTypeAndUploadInstantly(String key, File file) {
        //filter the file type and upload instantly
        if (FILE_ID_LIST_TO_INSTANT_UPLOAD.contains(key)) {
            try {
                String remotePath = assetService.getSystemProperty("DDS_SFTP_REMOTE_PATH");
                String localBackUpPath = assetService.getSystemProperty("DDS_SFTP_LOCALDIR");
                remotePath = StrUtil.isNotEmpty(remotePath) ? remotePath : sftpConfig.getRemotePath();
                localBackUpPath = StrUtil.isNotEmpty(localBackUpPath) ? localBackUpPath : sftpConfig.getLocalDir();
                sftpUtils.upLoadSingleFile(remotePath, sftpConfig, file, localBackUpPath);
            } catch (SftpException | IOException e) {
                log.error("SftpException:[{}]", e.toString());
            }
        }
    }

    public Map<Integer, List<String>> transformDdsEventForMdm(List<TaskEventData> taskEventDatas, Integer fileType) {
        log.debug("start dds file generation for event, the request : {}", JSONUtil.toJsonStr(taskEventDatas));
        if (CollectionUtils.isEmpty(taskEventDatas)) {
            return null;
        }

        //用于生成最终文件
        Map<Integer, List<String>> result = new HashMap<>();
        //用于相同dataItemId的数据合并
        List<MdmEventDataDto> mdmEventDataDtoList = new ArrayList<>();
        List<RdEventMeter> rdEventMeterList = taskEventDatas.stream().filter(o -> !CollectionUtils.isEmpty(o.getRdEventMeter())).map(TaskEventData::getRdEventMeter).flatMap(Collection::stream).collect(Collectors.toList());
        List<RdEventDcu> rdEventDcuList = taskEventDatas.stream().filter(o -> !CollectionUtils.isEmpty(o.getRdEventDcu())).map(TaskEventData::getRdEventDcu).flatMap(Collection::stream).collect(Collectors.toList());

        List<String> missingEventMappingRuleList = new ArrayList<>();
        List<Long> missingEventDataIdList = new ArrayList<>();
        List<Long> missingMeterIdList = new ArrayList<>();
        List<Long> missingDcuIdList = new ArrayList<>();

        if (CollectionUtil.isNotEmpty(rdEventMeterList)) {
            for (RdEventMeter rdEvent : rdEventMeterList) {
                MdmEventDataDto mdmEventDataDto = new MdmEventDataDto();
                //parse meter event
                Long meterId = rdEvent.getMeterId();
                if (null == meterId) {
                    continue;
                }
                String eventTypeId = getMdmEventDataItemByType(String.valueOf(rdEvent.getEventType()));

                if (null == eventTypeId) {
                    continue;
                }
                mdmEventDataDto.setEventTypeId(eventTypeId);
                mdmEventDataDto.setEventCode(rdEvent.getEventCode());
                Meter meter = assetService.getMeterById(meterId);
                mdmEventDataDto.setAssetNo(meter.getCustomerSn());
                mdmEventDataDto.setAssetType(Math.toIntExact(DeviceType.METER.getId()));
                mdmEventDataDto.setTv(rdEvent.getTv());
                if (Long.valueOf(0).equals(rdEvent.getTvUpdate())) {
                    mdmEventDataDto.setTvUpdate(System.currentTimeMillis() / 1000);
                } else {
                    mdmEventDataDto.setTvUpdate(rdEvent.getTvUpdate());
                }
                mdmEventDataDto.setEventFlag(EVENT_FLAG_OCCUR);

                Map<Integer, String> mdmEventDataItemMap = new HashMap<>();
                if (CollectionUtil.isNotEmpty(rdEvent.getSubData())) {
                    for (RdEventMeterData rdEventMeterData : rdEvent.getSubData()) {
                        Long subDataItemId = rdEventMeterData.getSubDataItemId();
                        Integer eventDataItemTypeId = MdmEventDataItemEnum.getMdmDataItemId(subDataItemId);
                        if (null == eventDataItemTypeId) {
                            missingEventDataIdList.add(subDataItemId);
                            continue;
                        }

                        String value = getMDMValue(meter.getModelId(), subDataItemId, rdEventMeterData.getValue());
                        if (value == null) {
                            log.error("MDM convert data null, {}", JSONUtil.toJsonStr(rdEvent));
                            continue;
                        }
                        mdmEventDataItemMap.put(eventDataItemTypeId, value);
                    }
                }
                mdmEventDataDto.setSubDataItems(mdmEventDataItemMap);
                mdmEventDataDtoList.add(mdmEventDataDto);
                mdmEventDataDto.setSubDataItemCount(mdmEventDataItemMap.size());
            }
        }

        if (CollectionUtil.isNotEmpty(rdEventDcuList)) {
            for (RdEventDcu rdEventDcu : rdEventDcuList) {
                MdmEventDataDto mdmEventDataDto = new MdmEventDataDto();
                Long dcuId = rdEventDcu.getDcuId();
                //parse meter event
                if (null == dcuId) {
                    continue;
                }
                String eventTypeId = getMdmEventDataItemByType(String.valueOf(rdEventDcu.getEventType()));

                if (null == eventTypeId) {
                    continue;
                }
                mdmEventDataDto.setEventTypeId(eventTypeId);
                mdmEventDataDto.setEventCode(rdEventDcu.getEventCode());
                Dcu dcu = assetService.getDcuById(dcuId);
                mdmEventDataDto.setAssetNo(dcu.getCustomerSn());
                mdmEventDataDto.setAssetType(Integer.valueOf(Math.toIntExact(DeviceType.DCU.getId())));
                mdmEventDataDto.setTv(rdEventDcu.getTv());
                if (Long.valueOf(0).equals(rdEventDcu.getTvUpdate())) {
                    mdmEventDataDto.setTvUpdate(System.currentTimeMillis() / 1000);
                } else {
                    mdmEventDataDto.setTvUpdate(rdEventDcu.getTvUpdate());
                }
                mdmEventDataDto.setEventFlag(EVENT_FLAG_OCCUR);

                Map<Integer, String> mdmEventDataItemMap = new HashMap<>();
                if (CollectionUtil.isNotEmpty(rdEventDcu.getSubData())) {
                    for (RdEventDcuData rdEventDcuData : rdEventDcu.getSubData()) {
                        Long subDataItemId = rdEventDcuData.getSubDataItemId();
                        Integer eventDataItemTypeId = MdmEventDataItemEnum.getMdmDataItemId(subDataItemId);
                        if (null == eventDataItemTypeId) {
                            missingEventDataIdList.add(subDataItemId);
                            continue;
                        }
                        String value = getMDMValue(dcu.getModelId(), subDataItemId, rdEventDcuData.getValue());
                        if (value == null) {
                            log.error("MDM convert data null, {}", JSONUtil.toJsonStr(rdEventDcuData));
                            continue;
                        }
                        mdmEventDataItemMap.put(eventDataItemTypeId, value);
                    }
                }
                mdmEventDataDto.setSubDataItems(mdmEventDataItemMap);

                mdmEventDataDtoList.add(mdmEventDataDto);
                mdmEventDataDto.setSubDataItemCount(mdmEventDataItemMap.size());
            }
        }

        //如下MDM事件数据项待映射：6299648  11944448  11948288 11948544 11948800
        if (CollectionUtil.isNotEmpty(missingEventMappingRuleList)
                || CollectionUtil.isNotEmpty(missingMeterIdList)
                || CollectionUtil.isNotEmpty(missingDcuIdList)
                || CollectionUtil.isNotEmpty(missingEventDataIdList)) {
            log.debug("dds event : some records cannot be handled as the correct way," +
                            "the missingMappingRuleList is :{}, the missingMeterIdList is :{}" +
                            ", the missingDcuIdList is : {}" +
                            ", the missingEventDataId is : {}",
                    JSONUtil.toJsonStr(missingEventMappingRuleList),
                    JSONUtil.toJsonStr(missingMeterIdList),
                    JSONUtil.toJsonStr(missingDcuIdList),
                    JSONUtil.toJsonStr(missingEventDataIdList));
        }
        List<String> resultStringList = new ArrayList<>();
        if (fileType == 1) {
            //TXT
            resultStringList = mdmEventDataDtoList.stream().map(item -> item.toMdmFormatString()).collect(Collectors.toList());
        } else if (fileType == 2) {
            resultStringList = mdmEventDataDtoList.stream().map(item -> item.toDdsFormatString()).collect(Collectors.toList());
        }
        result.put(DdsConst.DDS_RD_TERM_ALARM_TABLE_ID, resultStringList);
        log.info("the built dds event file size is : {}", JSONUtil.toJsonStr(result));
        return result;
    }

    public Map<Integer, List<String>> transformDdsDeviceStatusForMdm(List<DeviceStatusDTO> deviceStatusDTOS) {
        if (CollectionUtils.isEmpty(deviceStatusDTOS)) {
            return null;
        }
        Map<Integer, List<String>> result = new HashMap<>();
        // 过滤出meter的和dcu的
        List<DeviceStatusDTO> meterDeviceStatusList = deviceStatusDTOS.stream().filter(item -> item.getDeviceType().equals(DeviceType.METER.getId())).collect(Collectors.toList());

        List<DeviceStatusDTO> dcuDeviceStatusList = deviceStatusDTOS.stream().filter(item -> item.getDeviceType().equals(DeviceType.DCU.getId())).collect(Collectors.toList());

        List<MdmDeviceStatusDto> meterDataDtoList = new ArrayList<>();
        for (DeviceStatusDTO dto : meterDeviceStatusList) {
            MdmDeviceStatusDto mdmDeviceStatusDto = new MdmDeviceStatusDto();
            mdmDeviceStatusDto.setTv(dto.getTv() == null ? SystemClock.now() / 1000 : dto.getTv());
            String customerSn = null;
            if (StrUtil.isNotEmpty(dto.getCustomerSn())) {
                customerSn = dto.getCustomerSn();
            } else {
                Meter meter = assetService.getMeterById(dto.getDeviceId());
                if (meter == null) {
                    log.error("the meter not found, meterId{}", dto.getDeviceId());
                    continue;
                }
                customerSn = meter.getCustomerSn();
            }
            mdmDeviceStatusDto.setCustomerSn(customerSn);
            mdmDeviceStatusDto.setDataSource(StrUtil.toString(MdmSourceEnum.ScheduleRead.getCode()));
            mdmDeviceStatusDto.setDataType(0);
            mdmDeviceStatusDto.setOnLineStatus(dto.getDeviceStatus());
            mdmDeviceStatusDto.setValidity(0);
            mdmDeviceStatusDto.setTvUpdate(SystemClock.now() / 1000);
            mdmDeviceStatusDto.setNetAddr(dto.getNetAddr());

            meterDataDtoList.add(mdmDeviceStatusDto);
        }

        List<MdmDeviceStatusDto> dcuDataDtoList = new ArrayList<>();
        for (DeviceStatusDTO dto : dcuDeviceStatusList) {
            MdmDeviceStatusDto mdmDeviceStatusDto = new MdmDeviceStatusDto();
            mdmDeviceStatusDto.setTv(dto.getTv() == null ? SystemClock.now() / 1000 : dto.getTv());
            String customerSn = null;
            if (StrUtil.isNotEmpty(dto.getCustomerSn())) {
                customerSn = dto.getCustomerSn();
            } else {
                Dcu dcu = assetService.getDcuById(dto.getDeviceId());
                if (dcu == null) {
                    log.error("the dcu not found, dcuId{}", dto.getDeviceId());
                    continue;
                }
                customerSn = dcu.getCustomerSn();
            }
            mdmDeviceStatusDto.setCustomerSn(customerSn);
            mdmDeviceStatusDto.setDataSource(StrUtil.toString(MdmSourceEnum.ScheduleRead.getCode()));
            mdmDeviceStatusDto.setDataType(0);
            mdmDeviceStatusDto.setOnLineStatus(dto.getDeviceStatus());
            mdmDeviceStatusDto.setValidity(0);
            mdmDeviceStatusDto.setTvUpdate(SystemClock.now() / 1000);
            mdmDeviceStatusDto.setNetAddr(dto.getNetAddr());

            dcuDataDtoList.add(mdmDeviceStatusDto);
        }

        List<String> resultDcuStringList = dcuDataDtoList.stream().map(MdmDeviceStatusDto::toMdmFormatString).collect(Collectors.toList());

        List<String> resultMeterStringList = meterDataDtoList.stream().map(MdmDeviceStatusDto::toMdmFormatString).collect(Collectors.toList());

        if (CollUtil.isNotEmpty(resultDcuStringList)) {
            result.put(DdsConst.DDS_HES_TERMINAL_STATE_TABLE_ID, resultDcuStringList);
        }
        if (CollUtil.isNotEmpty(resultMeterStringList)) {
            result.put(DdsConst.DDS_HES_METER_STATE_TABLE_ID, resultMeterStringList);
        }

        return result;
    }

    public Map<Integer, List<String>> transformDdsAnonyDeviceStatusForMdm(List<MdmAnonyTerminalDto> deviceStatusDTOS) {
        if (CollectionUtils.isEmpty(deviceStatusDTOS)) {
            return null;
        }
        Map<Integer, List<String>> result = new HashMap<>();

        List<String> resultDataList = deviceStatusDTOS.stream().map(MdmAnonyTerminalDto::toMdmFormatString).collect(Collectors.toList());

        if (CollUtil.isNotEmpty(resultDataList)) {
            result.put(DdsConst.DDS_HES_ANONY_TERMINAL_TABLE_ID, resultDataList);
        }


        return result;
    }

    private String getMdmEventDataItemByType(String eventTypeId) {
        return mappingEventRules.get(RedisKeys.getMdmDataTypeId(eventTypeId));
    }

    public void tmpFileCheck(String localDir) {
        List<String> fileNames = FileUtil.listFileNames(localDir);
        List<String> notInEngineTmpFileNames = fileNames.stream().filter(s -> s.endsWith(TMP)).filter(s -> !ddsFileMap.values().stream().filter(p -> s.startsWith(p.getFirst())).findAny().isPresent()).collect(Collectors.toList());
        log.info("dds tmp file not in Engine Process： {}", JSONUtil.toJsonStr(notInEngineTmpFileNames));
        notInEngineTmpFileNames.stream().forEach(s -> {
            String newFileName = s.replace(TMP, TXT);
            FileUtil.rename(FileUtil.file(localDir + s), newFileName, false, true);
        });

        List<Map.Entry<Integer, Pair<String, Long>>> overTimeFiles = ddsFileMap.entrySet().stream().filter(p -> SystemClock.now() - p.getValue().getSecond() > 3 * 60 * 1000).collect(Collectors.toList());

        if (!CollectionUtils.isEmpty(overTimeFiles)) {
            log.info("dds tmp file over 3~8 minutes: {}", JSONUtil.toJsonStr(overTimeFiles));
            synchronized (ddsFileMap) {
                overTimeFiles.stream().forEach(entry -> {
                    Pair<String, Long> p = entry.getValue();
                    String newFileName = StrUtil.concat(true, p.getFirst(), TXT);
                    FileUtil.rename(FileUtil.file(localDir + p.getFirst() + TMP), newFileName, false, true);
                    ddsFileMap.remove(entry.getKey());
                });
            }
        }
    }

    /**
     * 获取CSV文件名称
     *
     * @param dataType
     * @return
     */
    public String getCsvFileName(Integer dataType) {
        Date date = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
        String dataTime = simpleDateFormat.format(date);
        String fileName = null;
        switch (dataType) {
            case 1:
                fileName = DAILY + UNDER_LINE + dataTime;
                break;
            case 2:
                fileName = MONTHLY + UNDER_LINE + dataTime;
                break;
            case 3:
                fileName = YEARLY + UNDER_LINE + dataTime;

                break;
            case 4:
                fileName = EVENT + UNDER_LINE + dataTime;
                break;
            default:
                break;
        }
        return fileName;
    }

    /**
     * 创建CSV文件(年月日冻结曲线)并上传
     *
     * @param groupedRdMeterReads
     */
    public void createCsvFileAndUpload(Map<List<Object>, List<RdMeterReadsLp>> groupedRdMeterReads, String localFilePath) {
        //日曲线CSV表头
        List<Object> headDay = new ArrayList<>();
        headDay.add(METER_NO);
        headDay.add(DATE_TIME);

        //月曲线CSV表头
        List<Object> headMonth = new ArrayList<>();
        headMonth.add(METER_NO);
        headMonth.add(DATE_TIME);
        //CSV数据(日冻结和月冻结数据)
        List<List<Object>> dailyDataList = new ArrayList<>();
        List<List<Object>> monthlyDataList = new ArrayList<>();

        for (Map.Entry<List<Object>, List<RdMeterReadsLp>> lpReadInfoEntry : groupedRdMeterReads.entrySet()) {
            List<Object> lpReadInfo = lpReadInfoEntry.getKey();
            Long lpDataItemId = Convert.toLong(lpReadInfo.get(0));
            Long meterId = Convert.toLong(lpReadInfo.get(2));
            Long tv = Convert.toLong(lpReadInfo.get(3));
            Meter meter = assetService.getMeterById(meterId);
            if (meter == null) {
                log.error("dds file Meter not exists. meterId={}", meterId);
                continue;
            }

            //分钟曲线直接返回
            if (lpDataItemId.equals(MIN_LP)) {
                log.info("grouped dds data: lpDataItemId={}, mdmDataTypeId={}, meterId={}, tv={}", lpDataItemId, lpReadInfo.get(1), lpReadInfo.get(2), lpReadInfo.get(3));
                continue;
            }
            Integer mdmDataTypeId = Convert.toInt(lpReadInfo.get(1));
            if (mdmDataTypeId == -1) {
                continue;
            }

            Date date = new Date(tv * 1000);
            //日冻结曲线
            if (lpDataItemId.equals(DAY_LP)) {
                //拼接一行数据
                List<Object> dataOneRow = new ArrayList<>();
                dataOneRow.add(meter.getCustomerSn());
                SimpleDateFormat simpleDateFormat = new SimpleDateFormat(YYYYMMDD);
                String format = simpleDateFormat.format(date);
                dataOneRow.add(format);
                List<MapDdsMdmWf> mapDdsMdmWfs = mapDdsMdmWfMapper.getListByTpeId(mdmDataTypeId);
                if (null != mapDdsMdmWfs && mapDdsMdmWfs.size() > 0) {
                    for (int i = 0; i < mapDdsMdmWfs.size(); i++) {
                        MapDdsMdmWf mapDdsMdmWf = mapDdsMdmWfs.get(i);
                        if (headDay.size() != mapDdsMdmWfs.size() + 2) {
                            headDay.add(mapDdsMdmWf.getMdmDataItemName());
                        }
                        AtomicReference<String> value = new AtomicReference<>("");
                        List<RdMeterReadsLp> readsLps = groupedRdMeterReads.get(lpReadInfo);
                        if (null != readsLps && readsLps.size() > 0) {
                            readsLps.forEach(lp -> {
                                if (lp.getDataItemId().equals(mapDdsMdmWf.getSubDataItemId())) {
                                    //单位转换
                                    String valueMdm = getMDMValue(meter.getModelId(), lp.getDataItemId(), lp.getValue());
                                    value.set(valueMdm);
                                }
                            });
                        }
                        dataOneRow.add(value);
                    }
                    dailyDataList.add(dataOneRow);
                }
            }

            //月冻结曲线
            else if (lpDataItemId.equals(MONTH_LP)) {
                List<Object> dataOneRow = new ArrayList<>();
                dataOneRow.add(meter.getCustomerSn());
                SimpleDateFormat simpleDateFormat = new SimpleDateFormat(YYYYMM);
                String format = simpleDateFormat.format(date);
                dataOneRow.add(format);
                List<MapDdsMdmWf> mapDdsMdmWfs = mapDdsMdmWfMapper.getListByTpeId(mdmDataTypeId);
                if (null != mapDdsMdmWfs && mapDdsMdmWfs.size() > 0) {
                    for (int i = 0; i < mapDdsMdmWfs.size(); i++) {
                        MapDdsMdmWf mapDdsMdmWf = mapDdsMdmWfs.get(i);
                        if (headMonth.size() != mapDdsMdmWfs.size() + 2) {
                            headMonth.add(mapDdsMdmWf.getMdmDataItemName());
                        }
                        AtomicReference<String> value = new AtomicReference<>("");
                        List<RdMeterReadsLp> readsLps = groupedRdMeterReads.get(lpReadInfo);
                        if (null != readsLps && readsLps.size() > 0) {
                            readsLps.forEach(lp -> {
                                if (lp.getDataItemId().equals(mapDdsMdmWf.getSubDataItemId())) {
                                    //单位转换
                                    String valueMdm = getMDMValue(meter.getModelId(), lp.getDataItemId(), lp.getValue());
                                    value.set(valueMdm);
                                }
                            });
                        }
                        dataOneRow.add(value);
                    }
                    monthlyDataList.add(dataOneRow);
                }
            }
        }
        if (dailyDataList.size() > 0) {
            String fileName = this.getCsvFileName(1);
            log.debug("createCsvFileAndUpload fileName={},dailyDataList={}", fileName, JSONUtil.toJsonStr(dailyDataList));
            CsvFileUtils.createCsvFile(headDay, dailyDataList, localFilePath, fileName);
        }
        if (monthlyDataList.size() > 0) {
            String fileName = this.getCsvFileName(2);
            log.debug("createCsvFileAndUpload fileName={},monthlyDataList={}", fileName, JSONUtil.toJsonStr(dailyDataList));
            CsvFileUtils.createCsvFile(headMonth, monthlyDataList, localFilePath, fileName);
        }
    }

    /**
     * 创建CSV文件(事件)并上传
     *
     * @param mdmEventDatas
     * @param localFilePath
     * @return
     */
    public void createCsvEvenFileAndUpload(List<String> mdmEventDatas, String localFilePath) {
        //事件CSV表头
        List<Object> headEvent = new ArrayList<>();
        headEvent.add(METER_NO);
        headEvent.add(EVEN_CODE);
        headEvent.add(EVENT_OCCUR_TIME);

        //事件数据
        List<List<Object>> eventDataList = new ArrayList<>();
        mdmEventDatas.forEach(mdmEventData -> {
            List<Object> eventDataOneRow = new ArrayList<>();
            String[] s = mdmEventData.split(",");
            if (null != s && s.length == 3) {
                eventDataOneRow.add(s[0]);
                eventDataOneRow.add(s[1]);
                Long tv = Long.parseLong(s[2]);
                Date date = new Date(tv * 1000);
                SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
                String dataTime = simpleDateFormat.format(date);
                eventDataOneRow.add(dataTime);
                eventDataList.add(eventDataOneRow);
            }
        });

        if (eventDataList.size() > 0) {
            String fileName = this.getCsvFileName(4);
            log.debug("createCsvEvenFileAndUpload fileName={},mdmEventDatas={}", fileName, JSONUtil.toJsonStr(mdmEventDatas));
            CsvFileUtils.createCsvFile(headEvent, eventDataList, localFilePath, fileName);
        }
    }

    public void clearScalerCache() {
        subDataItemScaleUnit.invalidateAll();
        subDataItemCoverRate.invalidateAll();
        appUnitName.invalidateAll();
    }
}
